/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you under the Apache License, Version 2.0 (the            *
 * "License"); you may not use this file except in compliance   *
 * with the License.  You may obtain a copy of the License at   *
 *                                                              *
 *   http://www.apache.org/licenses/LICENSE-2.0                 *
 *                                                              *
 * Unless required by applicable law or agreed to in writing,   *
 * software distributed under the License is distributed on an  *
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
 * KIND, either express or implied.  See the License for the    *
 * specific language governing permissions and limitations      *
 * under the License.                                           *
 ****************************************************************/
package org.apache.james.mailbox.cassandra.mail;

import java.io.IOException;
import java.nio.charset.CharacterCodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.protocol.TBinaryProtocol;


import org.apache.james.mailbox.MailboxException;
import org.apache.james.mailbox.MailboxNotFoundException;
import org.apache.james.mailbox.MailboxPath;
import org.apache.james.mailbox.store.mail.MailboxMapper;
import org.apache.james.mailbox.store.mail.model.Mailbox;
import java.nio.ByteBuffer;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.utils.ByteBufferUtil;

import org.apache.james.mailbox.cassandra.CNames;
import org.apache.james.mailbox.cassandra.CNonTransactionalMapper;
import org.apache.james.mailbox.cassandra.CUtils;
import org.apache.james.mailbox.cassandra.mail.model.CMailbox;


/**
 * Data access management for mailbox.
 * 
 */
public class CMailboxMapper extends CNonTransactionalMapper implements MailboxMapper<UUID> {

        	
         
    @Override
    public Mailbox<UUID> findMailboxByPath(MailboxPath mailboxPath) throws MailboxException, MailboxNotFoundException {
      
        return null;
    }

    @Override
    public List<Mailbox<UUID>> findMailboxWithPathLike(MailboxPath mailboxPath) throws MailboxException {
       
        return null;
    }

    @Override
    public List<Mailbox<UUID>> list() throws MailboxException {
       
        try {
            
            
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         
         transport.open();
         
         
         Cassandra.Client client = new Cassandra.Client(protocol);
         client.set_keyspace(CNames.keyspace);  
            
         ColumnParent mailboxes= new ColumnParent(ByteBufferUtil.string(CNames.MAILBOX_CF)); 
         List<Mailbox<UUID>> mailboxList = new ArrayList<Mailbox<UUID>>();
         Mailbox m;
            
         SlicePredicate predicate = new SlicePredicate();
         predicate.setSlice_range(new SliceRange(ByteBuffer.wrap(new byte[0]), ByteBuffer.wrap(new byte[0]), false, 100));
         KeyRange keyRange = new KeyRange(100);
         keyRange.setStart_key(new byte[0]);
         keyRange.setEnd_key(new byte[0]);
         List<KeySlice> keySlices = client.get_range_slices(mailboxes, predicate, keyRange, ConsistencyLevel.ONE);
         for (KeySlice ks : keySlices) 
         {
             
             //*************
             
                m = CUtils.mailboxFromResult((ks.getKey()).toString(), CNames.keyspace);
                mailboxList.add(m);
         }
         return mailboxList;
         } catch (IOException ex) {
            throw new MailboxException("HBase IOException in list()", ex);
         } finally {
            
                    throw new MailboxException("Error opening columnfamily ");
                
            }
        }
    

    @Override
    public void endRequest() {
    }

    @Override
    public void save(Mailbox<UUID> mlbx) throws MailboxException {
        //TODO: maybe switch to checkAndPut for transactions
      //  HTable mailboxes = null;
        
         ColumnParent mailboxes = null;
         
         try {
        
        
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         
         transport.open();
         
         
         Cassandra.Client client = new Cassandra.Client(protocol);
         client.set_keyspace(CNames.keyspace);  
         
         mailboxes= new ColumnParent(ByteBufferUtil.string(CNames.MAILBOX_CF));
        
         client = CUtils.toPut((CMailbox)mlbx);
            
        } catch (TTransportException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (CharacterCodingException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException ex) {
            throw new MailboxException("IOExeption", ex);
        } finally {
            if (mailboxes == null)
              throw new MailboxException("Error opening columnfamily " );
            }
        }
    

    @Override
    public void delete(Mailbox<UUID> mlbx) throws MailboxException  {
        //TODO: maybe switch to checkAndDelete
       // HTable mailboxes = null;
        byte [] rowkey=null;
        
        try {
            
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         
         transport.open();
         
         
         Cassandra.Client client = new Cassandra.Client(protocol);
         client.set_keyspace(CNames.keyspace);  
           //TODO: delete all maessages from this mailbox
            //Delete delete = new Delete(mailboxRowKey(mlbx.getMailboxId()));
         rowkey = CUtils.mailboxRowKey(mlbx.getMailboxId());
         
         //************
         
         CqlResult execute_cql_query = client.execute_cql_query(ByteBufferUtil.bytes("DELETE FROM MAILBOXES_TABLE WHERE KEY IN ('" + (rowkey).toString() + "')"), Compression.GZIP);
         } catch (InvalidRequestException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        
        } catch (TException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        } finally {
            
                    throw new MailboxException("Error closing table ");  
        }
    }

    @Override
    public boolean hasChildren(final Mailbox<UUID> mailbox, final char c) throws MailboxException, MailboxNotFoundException {
     
        return false;
    }

    public void deleteAllMemberships() {
      
    }

    public void deleteAllMailboxes() {
        
         try {
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         
         transport.open();
         
         
         Cassandra.Client client = new Cassandra.Client(protocol);
         client.set_keyspace(CNames.keyspace);  
           
                //HTable mailboxes = null;
               // ResultScanner scanner = null;
            client.truncate(ByteBufferUtil.string(CNames.MAILBOXES_TABLE));
            } catch (InvalidRequestException ex) {
                Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
            } catch (CharacterCodingException ex) {
            Logger.getLogger(CMailboxMapper.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}
